package com.swak.vertx.invoker;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.asm.MethodCache.MethodMeta;
import com.swak.exception.BlockException;
import com.swak.reactivex.threads.BlockedThreadChecker;
import com.swak.reactivex.threads.BlockedThreadChecker.Task;
import com.swak.reactivex.threads.Context;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.utils.StringUtils;
import com.swak.vertx.Vertxs;
import com.swak.vertx.transport.codec.Msg;
import com.swak.vertx.transport.script.Script;

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.Message;

/**
 * 消息的异步处理
 *
 * @author: lifeng
 * @date: 2020/3/29 19:43
 */
public interface FluxInvoker {

	/**
	 * 通过类型解析出服务地址
	 *
	 * @param type 类型
	 * @return 服务地址
	 */
	default String getAddress(Class<?> type, String address) {

		/*
		 * 默认使用接口的全额限定名称
		 */
		if (StringUtils.isBlank(address)) {
			address = type.getName();
		}

		/*
		 * 约定去掉后面的 Async
		 */
		return StringUtils.removeEnd(address, Constants.ASYNC_SUFFIX);
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, Script script) {

		// 构建异步请求结果
		TimeoutPromise promise = new TimeoutPromise(address, -1);

		// 发送消息，处理相应结果
		this.sendRemoteMessage(address, new Msg(script), -1, promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, MethodMeta meta, Object[] args) {

		// 构建异步请求结果
		TimeoutPromise promise = new TimeoutPromise(address, meta.getTimeOut());

		// 发送消息，处理相应结果
		// Vertxs.sentMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);
		this.sendRemoteMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, MethodMeta meta, Object[] args,
			boolean callBackInSourceContext) {

		// 构建异步请求结果
		TimeoutPromise promise = new TimeoutPromise(address, meta.getTimeOut(), callBackInSourceContext);

		// 发送消息，处理相应结果
		// Vertxs.sentMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);
		this.sendRemoteMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	default void sendRemoteMessage(String address, Msg request, int timeout, TimeoutPromise promise) {
		try {
			Vertxs.sentMessage(address, request, timeout, promise);
		} catch (Throwable e) {
			promise.finish(null, e);
		}
	}

	/**
	 * 减少对象的创建
	 * 
	 * @author DELL
	 */
	static class TimeoutPromise extends ForbidBlockCompletableFuture<Object>
			implements Handler<AsyncResult<Message<Msg>>>, Task {

		private static final Logger log = LoggerFactory.getLogger(BlockedThreadChecker.class);
		private static final Map<Long, TimeoutPromise> FUTURES = new ConcurrentHashMap<>();
		private static final AtomicLong PROMISE_ID = new AtomicLong(1);
		private static final Integer WarningExceptionTimeMs = 60 * 1000;
		private static final Integer ErrorExceptionTimeMs = 3 * 60 * 1000;
		private static final ScheduledExecutorService TIME_OUT_TIMER = new ScheduledThreadPoolExecutor(1,
				new SwakThreadFactory("Flux-blocked-Future-checker.", true, new AtomicInteger()));;
		static {
			TIME_OUT_TIMER.scheduleAtFixedRate(TimeoutPromise::checkBlockedFuture, 60, 60, TimeUnit.SECONDS);
		}

		static void checkBlockedFuture() {
			if (FUTURES.isEmpty()) {
				return;
			}
			synchronized (FUTURES) {
				long now = System.nanoTime();
				for (Map.Entry<Long, TimeoutPromise> entry : FUTURES.entrySet()) {
					long execStart = entry.getValue().startTime();
					long dur = now - execStart;
					final long timeLimit = entry.getValue().maxExecTime();
					TimeUnit maxExecTimeUnit = entry.getValue().maxExecTimeUnit();
					long val = maxExecTimeUnit.convert(dur, TimeUnit.NANOSECONDS);
					if (val >= timeLimit) {
						final String message = "Future for[" + entry.getValue().address + "] has been blocked for "
								+ (dur / 1_000_000) + " ms, time limit is "
								+ TimeUnit.MILLISECONDS.convert(timeLimit, maxExecTimeUnit) + " ms";
						if (TimeUnit.MILLISECONDS.convert(dur, TimeUnit.NANOSECONDS) <= ErrorExceptionTimeMs) {
							log.debug(message);
						} else {
							BlockException stackTrace = new BlockException("Future blocked");
							log.warn(message, stackTrace);
						}
					}
				}
			}
		}

		final Long id;
		final String address;
		Context context = null;
		final long execStart;
		final long maxExecTime;
		final TimeUnit maxExecTimeUnit;

		TimeoutPromise(String address, long timeout) {
			this(address, timeout, false);
		}

		TimeoutPromise(String address, long timeout, boolean callBackInSourceContext) {
			this.id = PROMISE_ID.getAndIncrement();
			this.address = address;
			this.execStart = System.nanoTime();
			this.maxExecTimeUnit = TimeUnit.MILLISECONDS;
			this.maxExecTime = timeout > 0 ? timeout : WarningExceptionTimeMs;
			if (callBackInSourceContext) {
				this.context = Contexts.currentContext();
			}

			// put into waiting map.
			FUTURES.put(this.id, this);
		}

		@Override
		public long startTime() {
			return execStart;
		}

		@Override
		public long maxExecTime() {
			return maxExecTime;
		}

		@Override
		public TimeUnit maxExecTimeUnit() {
			return maxExecTimeUnit;
		}

		/**
		 * 这个返回一定会在 Vertx 的上下文中执行, 如果在Context中执行，则进行切换
		 */
		@Override
		public void handle(AsyncResult<Message<Msg>> event) {
			if (context != null) {
				context.runOnContext(() -> this.handleResult(event));
				return;
			}
			this.handleResult(event);
		}

		private void handleResult(AsyncResult<Message<Msg>> event) {

			// 错误 和输出
			Throwable result_error = null;
			Object result_result = null;

			// 结果处理
			try {
				// 内部处理异常
				if (event.cause() != null) {
					result_error = event.cause();
				}

				// 处理正确
				else {
					// 约定的通讯协议
					Msg result = event.result().body();

					// 错误处理 - 结果返回
					result_error = result.getError();
					result_result = result.getResult();
				}
			} catch (Throwable e) {
				result_error = e;
			}

			// 结束当前的Future
			this.finish(result_result, result_error);
		}

		public void finish(Object result_result, Throwable result_error) {
			try {
				// 优先错误处理
				if (result_error != null) {
					this.completeExceptionally(result_error);
				}

				// 结果返回
				else {
					this.complete(result_result);
				}
			} finally {
				FUTURES.remove(this.id);
			}
		}
	}
}